package net.bwie.realtime.function;

import net.bwie.realtime.dws.log.bean.MonitorInfo;
import net.bwie.realtime.dws.log.bean.Violation;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

public class TaoPaiRichFlatMapFunction extends RichFlatMapFunction<MonitorInfo, Violation> {

    //step1.声明状态
    ValueState<MonitorInfo>  valueState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // step2.初始化状态
        valueState = getRuntimeContext().getState(
                new ValueStateDescriptor<MonitorInfo>("vs1", MonitorInfo.class)
        );
    }

    @Override
    public void flatMap(MonitorInfo value, Collector<Violation> out) throws Exception {
        // 根据车牌号，获取上一个汽车的 时间，还有 摄像头的id
        MonitorInfo speedInfo = valueState.value();
        valueState.update(speedInfo);
        if (speedInfo != null && value.getActionTime() - speedInfo.getActionTime() < 10 && value.getMonitorId() != speedInfo.getMonitorId()) {
            Violation violation = new Violation(0, speedInfo.getCar(), "涉嫌套牌", System.currentTimeMillis());
            out.collect(violation);
        }
    }
}
